package com.atguigu.day08;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Example6 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var stream = env
                .addSource(new SourceFunction<Tuple3<String, Double, Long>>() {
                    @Override
                    public void run(SourceContext<Tuple3<String, Double, Long>> ctx) {
                        ctx.collectWithTimestamp(Tuple3.of("credit-card-1", 0.01, 1000L), 1000L);
                        ctx.collectWithTimestamp(Tuple3.of("credit-card-1", 1001.01, 2000L), 2000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        var pattern = Pattern
                .<Tuple3<String, Double, Long>>begin("first")
                .where(new SimpleCondition<>() {
                    @Override
                    public boolean filter(Tuple3<String, Double, Long> value) {
                        return value.f1 < 1.0;
                    }
                })
                .next("second")
                .where(new SimpleCondition<>() {
                    @Override
                    public boolean filter(Tuple3<String, Double, Long> value) {
                        return value.f1 > 1000.0;
                    }
                });

        CEP
                .pattern(stream.keyBy(r -> r.f0), pattern)
                .select(map -> {
                    var first = map.get("first").get(0);
                    var second = map.get("second").get(0);
                    return first.f0 + "可能被盗，连续两笔消费是：" + first.f1 + ", " + second.f1;
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
